package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TrafficHomeDetailPageViewBean;
import com.atguigu.gmall.realtime.util.ClickHouseUtil;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author caodan
 * @version 1.0
 * @date 2022-10-11 9:16
 * 流量域首页、详情页页面浏览各窗口轻度聚合
 */
public class DwsTrafficHomeDetailPageViewWindow {

    public static void main(String[] args) throws Exception {

        // 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //检查点相关设置(略)

        // 从kafka中获取数据
        String topicName = "dwd_traffic_page_log";
        String groupId = "dws_traffic_detail_page_view_group";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtil.getFlinkKafkaConsumer(topicName, groupId);
        // 转换为流
        DataStreamSource<String> streamSourceDS = env.addSource(flinkKafkaConsumer);

        // 将jsonStr ---> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = streamSourceDS.map(JSON::parseObject);

        // 过滤元素，提取home和good_detail数据
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObj) throws Exception {
                String pageId = jsonObj.getJSONObject("page").getString("page_id");
                return "home".equals(pageId) || "good_detail".equals(pageId);
            }
        });

        // filterDS.print();

        // 插入水位线
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = filterDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                }
                        )
        );

        // 通过mid 进行分组
        KeyedStream<JSONObject, String> keyedStream = withWatermarkDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        // 使用状态编程记录
        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> processDS = keyedStream.process(new KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>() {
            ValueState<String> homeVisitValueSate;
            ValueState<String> detailVisitValueSate;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> homeVisitValueSate = new ValueStateDescriptor<>("homeVisitValueSate", String.class);
                homeVisitValueSate.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build()); // 设置状态过期时间
                this.homeVisitValueSate = getRuntimeContext().getState(homeVisitValueSate);

                ValueStateDescriptor<String> detailVisitValueSate = new ValueStateDescriptor<>("detailVisitValueSate", String.class);
                detailVisitValueSate.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                this.detailVisitValueSate = getRuntimeContext().getState(detailVisitValueSate);
            }

            @Override
            public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, TrafficHomeDetailPageViewBean>.Context ctx, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                String homeLastDt = homeVisitValueSate.value();
                String detailLastDt = detailVisitValueSate.value();

                JSONObject page = jsonObj.getJSONObject("page");
                String pageId = page.getString("page_id");
                Long ts = jsonObj.getLong("ts");
                String visitDt = DateFormatUtil.toDate(ts);

                long homeUvCt = 0L;
                long detailUvCt = 0L;

                if (pageId.equals("home")) {
                    if (homeLastDt == null || !homeLastDt.equals(visitDt)) {
                        homeUvCt = 1L;
                        homeVisitValueSate.update(visitDt);
                    }
                }

                if (pageId.equals("good_detail")) {
                    if (detailLastDt == null || !detailLastDt.equals(visitDt)) {
                        detailUvCt = 1L;
                        detailVisitValueSate.update(visitDt);
                    }
                }

                if (homeUvCt != 0 || detailUvCt != 0) {
                    out.collect(new TrafficHomeDetailPageViewBean(
                            "",
                            "",
                            homeUvCt,
                            detailUvCt,
                            0L
                    ));
                }
            }
        });

        SingleOutputStreamOperator<TrafficHomeDetailPageViewBean> reduce = processDS.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                .reduce(new ReduceFunction<TrafficHomeDetailPageViewBean>() {
                    @Override
                    public TrafficHomeDetailPageViewBean reduce(TrafficHomeDetailPageViewBean value1, TrafficHomeDetailPageViewBean value2) throws Exception {
                        value1.setGoodDetailUvCt(
                                value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt()
                        );
                        value1.setHomeUvCt(
                                value1.getHomeUvCt() + value2.getHomeUvCt()
                        );
                        return value1;
                    }
                }, new AllWindowFunction<TrafficHomeDetailPageViewBean, TrafficHomeDetailPageViewBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TrafficHomeDetailPageViewBean> values, Collector<TrafficHomeDetailPageViewBean> out) throws Exception {
                        for (TrafficHomeDetailPageViewBean trafficHomeDetailPageViewBean : values) {
                            trafficHomeDetailPageViewBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                            trafficHomeDetailPageViewBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                            trafficHomeDetailPageViewBean.setTs(System.currentTimeMillis());
                            out.collect(trafficHomeDetailPageViewBean);
                        }
                    }
                });

        // 将结果写入到 clickhouse 中
        reduce.addSink(
                ClickHouseUtil.getJdbcSink("insert into dws_traffic_home_detail_page_view_window values(?,?,?,?,?)")
        );

        env.execute();
    }
}
